package com.ouc.pampas;
import com.alibaba.dubbo.config.ApplicationConfig;
import com.alibaba.dubbo.config.ProtocolConfig;
import com.alibaba.dubbo.config.RegistryConfig;
import com.alibaba.dubbo.config.spring.ServiceBean;
import com.fasterxml.jackson.databind.JavaType;
import com.google.common.base.Charsets;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;
import io.terminus.common.utils.JsonMapper;
import io.terminus.pampas.engine.utils.ZkPaths;
import java.net.InetAddress;
import java.util.Map;
import javax.annotation.PostConstruct;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkMarshallingError;
import org.I0Itec.zkclient.serialize.ZkSerializer;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class Register
{
  @Autowired
  private ApplicationContext applicationContext;
  @Autowired
  private Setting setting;
  private ZkClient zkClient;
  private String groupName;
  private RegistryConfig registryConfig = null;
  
  @PostConstruct
  private void init()
    throws Exception
  {
    ApplicationConfig applicationConfig = (ApplicationConfig)this.applicationContext.getBean(ApplicationConfig.class);
    this.groupName = applicationConfig.getName();
    
    Preconditions.checkNotNull(this.groupName, "applicationConfig.getName() should not be null");
    final ProtocolConfig protocolConfig = (ProtocolConfig)this.applicationContext.getBean(ProtocolConfig.class);
    if ((!Strings.isNullOrEmpty(protocolConfig.getName())) && (!Objects.equal(protocolConfig.getName().toLowerCase(), "dubbo"))) {
      throw new IllegalArgumentException("only support dubbo protocol");
    }
    if ((protocolConfig.getPort() == null) || (protocolConfig.getPort().intValue() <= 0)) {
      throw new IllegalArgumentException("only support clearly port definition in dubbo protocol");
    }
    prepareZkClient();
    String groupPath = ZkPaths.groupPath(this.groupName);
    if (!this.zkClient.exists(groupPath)) {
      this.zkClient.createPersistent(groupPath, true);
    }
    String cellName = InetAddress.getLocalHost().getHostAddress();
    final String cellPath = ZkPaths.cellPath(this.groupName, cellName);
    IZkStateListener stateListener = new IZkStateListener()
    {
      public void handleNewSession()
        throws Exception
      {
        if (Register.this.zkClient.exists(cellPath)) {
          Register.this.zkClient.delete(cellPath);
        }
        Map<String, Object> cellInfo = Maps.newHashMap();
        cellInfo.put("port", protocolConfig.getPort());
        Register.this.zkClient.create(cellPath, cellInfo, CreateMode.EPHEMERAL);
      }

	@Override
	public void handleStateChanged(KeeperState state) throws Exception {
		// TODO 自动生成的方法存根
		
	}
    };
    this.zkClient.subscribeStateChanges(stateListener);
    
    stateListener.handleNewSession();
  }
  
  public <T> ServiceBean<T> createServiceBean(Class<T> clazz)
  {
    ServiceBean<T> serviceBean = new ServiceBean();
    if (this.registryConfig != null) {
      serviceBean.setRegistry(this.registryConfig);
    }
    serviceBean.setInterface(clazz);
    serviceBean.setVersion(this.groupName);
    serviceBean.setRef(this.applicationContext.getBean(clazz));
    return serviceBean;
  }
  
  @Bean
  public ServiceBean<ConfigService> dubboConfigService()
  {
    return createServiceBean(ConfigService.class);
  }
  
  @Bean
  public ServiceBean<AppService> dubboAppService()
  {
    return createServiceBean(AppService.class);
  }
  
  private void prepareZkClient()
  {
    Map<String, RegistryConfig> registryConfigs = this.applicationContext.getBeansOfType(RegistryConfig.class);
    if (registryConfigs.size() == 1)
    {
      this.registryConfig = ((RegistryConfig)registryConfigs.values().iterator().next());
    }
    else
    {
      if (Strings.isNullOrEmpty(this.setting.getRegistryId())) {
        throw new IllegalArgumentException("registryId in [Setting] should not be empty when multi dubbo registry configs");
      }
      if (!registryConfigs.containsKey(this.setting.getRegistryId())) {
        throw new IllegalArgumentException("no specific dubbo registry found for id: " + this.setting.getRegistryId());
      }
      this.registryConfig = ((RegistryConfig)registryConfigs.get(this.setting.getRegistryId()));
    }
    String zkUrl;
    if (Protocol.analyze(this.registryConfig.getAddress()) == Protocol.ZK)
    {
      zkUrl = Protocol.removeProtocol(this.registryConfig.getAddress(), Protocol.ZK);
    }
    else
    {
      if ("zookeeper".equalsIgnoreCase(this.registryConfig.getProtocol())) {
        zkUrl = this.registryConfig.getAddress();
      } else {
        throw new IllegalArgumentException("only supported zookeeper dubbo registry protocol");
      }
    }
    this.zkClient = new ZkClient(zkUrl);
    this.zkClient.setZkSerializer(new ZkSerializer()
    {
      private final JavaType mapType = JsonMapper.nonEmptyMapper().createCollectionType(Map.class, new Class[] { String.class, Object.class });
      
      public byte[] serialize(Object data)
        throws ZkMarshallingError
      {
        return JsonMapper.nonEmptyMapper().toJson(data).getBytes(Charsets.UTF_8);
      }
      
      public Object deserialize(byte[] bytes)
        throws ZkMarshallingError
      {
        return JsonMapper.nonEmptyMapper().fromJson(new String(bytes, Charsets.UTF_8), this.mapType);
      }
    });
  }
}
